package com.learn.kafka.assign;


import com.learn.kafka.config.ConsumerProperties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;

/**
 * 手动指定分区消费的消费者（继承通用配置父类）
 * @author: learn.kafka
 */
public class ConsumerHandAssignTest extends ConsumerProperties {


    public static void main(String[] args) {
        new ConsumerHandAssignTest().startConsuming();
    }

    // 构造方法：初始化手动分区分配的特殊配置
    public ConsumerHandAssignTest() {
        super(); // 调用父类构造方法，加载通用配置
        initHandAssignConfig();
    }

    // 初始化手动分区分配的配置（覆盖父类默认值）
    private void initHandAssignConfig() {
        // 设置消费组ID（覆盖父类默认的"default-group"）
        setProperty("group.id", "mate-assgin-group");
        // 如需修改其他配置（如自动提交间隔），可在此处添加
        // setConfig("auto.commit.interval.ms", "2000");
    }

    // 启动消费逻辑：手动指定分区消费
    private void startConsuming() {
        // 从父类获取消费者实例
        Consumer<Integer, String> consumer = createConsumer();

        // 手动指定消费的分区（主题"ass5part"的分区0）
        TopicPartition targetPartition = new TopicPartition("ass5part", 0);
        consumer.assign(Arrays.asList(targetPartition));

        try {
            // 持续消费指定分区的消息
            while (true) {
                ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<Integer, String> record : records) {
                    System.out.printf(
                            "offset = %d, key = %s, value = %s, partition = %d%n",
                            record.offset(), record.key(), record.value(), record.partition()
                    );
                }
            }
        } finally {
            // 关闭消费者，释放资源
            consumer.close();
        }
    }
}
